package com.devilvan.betrice.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author Evad.Wu
 * @Description 消息队列 控制类
 * @date 2023-02-10
 */
@Slf4j
@RestController
@RequestMapping(value = "betriceMqController")
public class BetriceMqController {
    @Resource(name = "betriceKafkaProducer")
    private Producer<String, String> producer;
    @Resource
    private StreamBridge streamBridge;

    @PostMapping("send")
    public void send(String topic, String key, String message) {
        try {
            producer.send(new ProducerRecord<>(topic, key, message));
            log.info("发送消息：" + message);
        } catch (Exception e) {
            log.error("异常消息：" + e);
        }
    }

    @PostMapping("streamSend")
    public void streamSend(String topic, String message) {
        try {
            streamBridge.send(topic, message);
            log.info("发送消息：" + message);
        } catch (Exception e) {
            log.error("异常消息：" + e);
        }
    }

    @PostMapping("sendGatherMessage")
    public Object sendGatherMessage(String mess1, String mess2) {
        streamBridge.send("gather1", mess1);
        streamBridge.send("gather2", mess2);
        return "Gather Message sended :" + mess1 + "; " + mess2;
    }
}
